package cn.itcast.map;

import cn.itcast.bean.IndexBean;
import cn.itcast.config.QuotConfig;
import cn.itcast.constant.Constant;
import cn.itcast.util.DateUtil;
import org.apache.flink.api.common.functions.MapFunction;

import java.sql.Timestamp;

public class IndexPutHdfsMap implements MapFunction<IndexBean,String> {
   private String sp = QuotConfig.config.getProperty("hdfs.seperator");

    @Override
    public String map(IndexBean indexBean) throws Exception {
        //Timestamp|date|indexCode|indexName|preClosePrice|openPirce|highPrice|
        //lowPrice|closePrice|tradeVol|tradeAmt|tradeVolDay|tradeAmtDay|source
        String date = DateUtil.longTimeToString(indexBean.getEventTime(), Constant.format_yyyy_mm_dd);
        StringBuilder sb = new StringBuilder();

        return sb.append(new Timestamp(indexBean.getEventTime())).append(sp)
                .append(date).append(sp)
                .append(indexBean.getIndexCode()).append(sp)
                .append(indexBean.getIndexName()).append(sp)
                .append(indexBean.getPreClosePrice()).append(sp)
                .append(indexBean.getOpenPrice()).append(sp)
                .append(indexBean.getHighPrice()).append(sp)
                .append(indexBean.getLowPrice()).append(sp)
                .append(indexBean.getClosePrice()).append(sp)
                .append(indexBean.getTradeVol()).append(sp)
                .append(indexBean.getTradeAmt()).append(sp)
                .append(indexBean.getTradeVolDay()).append(sp)
                .append(indexBean.getTradeAmtDay()).append(sp)
                .append(indexBean.getSource()).toString();
    }
}
